package cn.doitedu.hbase.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;

import java.io.IOException;

public class DemoMr {

    public static class M extends TableMapper<Text, IntWritable>{

        @Override
        protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
            String k = new Text(new String(key.copyBytes())).toString();
            String v = Bytes.toString(value.getValue("f".getBytes(),"v".getBytes()));

            System.out.println(k + " " + v);

            context.write(new Text(k),new IntWritable(Integer.parseInt(v)));
        }
    }


    public static class R extends TableReducer<Text,IntWritable, ImmutableBytesWritable>{

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            Put put = new Put(key.getBytes());
            put.addColumn("f".getBytes(),"v".getBytes(),Bytes.toBytes(values.iterator().next().get()));
            context.write(null,put);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","spark01,spark02,spark03");

        Scan scan = new Scan();
        scan.setCaching(500);
        scan.setCacheBlocks(false);


        Job job = Job.getInstance(conf);
        job.setJarByClass(DemoMr.class);
        TableMapReduceUtil.initTableMapperJob("t1",scan,M.class,Text.class,IntWritable.class,job);

        TableMapReduceUtil.initTableReducerJob("t2",R.class,job);

        boolean b = job.waitForCompletion(true);
        if(b) System.exit(1);
    }
}
